package com.zhaosc.rx.http;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

public class RxHttpClient {

	public  static void doGet(String url) {

		Observable.create(new ObservableOnSubscribe<String>() {
			@Override
			public void subscribe(ObservableEmitter<String> emitter) throws Exception {
				emitter.onNext(url);
			}
		}).map(new Function<String, HttpResponse>() {

			@Override
			public HttpResponse apply(String url) throws Exception {
				CloseableHttpClient client = HttpClients.createDefault();
				HttpGet get = new HttpGet(url);
				CloseableHttpResponse response = client.execute(get);
				return response;
			}
		})
		.subscribe(new Consumer<HttpResponse>() {

			@Override
			public void accept(HttpResponse response) throws Exception {
				HttpEntity entity=response.getEntity();
				if(response.getStatusLine().getStatusCode()==HttpStatus.SC_OK) {
					String result=EntityUtils.toString(entity);
					System.out.println(result);
				}
				EntityUtils.consume(entity);
			}
		});
	}

}
